﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.IO;
using System.Data;
using System.Reflection;
using System.Threading.Tasks;
using Fast.Framework.Interfaces;
using Fast.Framework.Extensions;


namespace Fast.Framework
{

    /// <summary>
    /// 大数据实现类
    /// </summary>
    public class BigDataProvider: IBigData
    {

        /// <summary>
        /// Ado
        /// </summary>
        private readonly IAdo ado;

        /// <summary>
        /// 构造方法
        /// </summary>
        /// <param name="ado">Ado</param>
        public BigDataProvider(IAdo ado)
        {
            this.ado = ado;
        }

        /// <summary>
        /// 批量复制
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="source">源</param>
        /// <param name="tableName">表名称</param>
        /// <param name="bulkCopyTimeout">批量复制超时 单位秒</param>
        /// <returns></returns>
        public Task<int> BulkCopyAsync<T>(IEnumerable<T> source, string tableName = "", int bulkCopyTimeout = 120) where T : class
        {
            return BulkCopyAsync(source.ToDataTable(tableName));
        }

        /// <summary>
        /// 批量复制
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="source">源</param>
        /// <param name="bulkCopyTimeout">批量复制超时 单位秒</param>
        /// <returns></returns>
        public async Task<int> BulkCopyAsync(DataTable dataTable, int bulkCopyTimeout = 120)
        {
            if (dataTable == null || dataTable.Rows.Count == 0)
            {
                throw new Exception("DataTable为Null或行数为0");
            }
            var conn = ado.DbProviderFactory.CreateConnection();
            conn.ConnectionString = ado.DbOptions.ConnectionStrings;

            #region SqlServer
            if (ado.DbOptions.DbType == Models.DbType.SQLServer)
            {

                var sqlClient = Assembly.LoadFrom(Path.Combine(AppContext.BaseDirectory, $"{ado.DbOptions.ProviderName}.dll"));
                await conn.OpenAsync();
                var tran = await conn.BeginTransactionAsync();

                var bulkCopy = Activator.CreateInstance(sqlClient.GetType($"{ado.DbOptions.ProviderName}.SqlBulkCopy"), new object[]
                {
                    conn,
                    Enum.Parse(sqlClient.GetType($"{ado.DbOptions.ProviderName}.SqlBulkCopyOptions"), "KeepIdentity"),
                    tran
                });

                var bulkCopyType = bulkCopy.GetType();

                bulkCopyType.GetProperty("DestinationTableName").SetValue(bulkCopy, dataTable.TableName);//表名
                bulkCopyType.GetProperty("BulkCopyTimeout").SetValue(bulkCopy, bulkCopyTimeout);

                var columnMappings = bulkCopyType.GetProperty("ColumnMappings").GetValue(bulkCopy);
                var columnMappingsAddMethod = bulkCopyType.GetProperty("ColumnMappings").PropertyType.GetMethod("Add", new Type[] { typeof(string), typeof(string) });
                foreach (DataColumn item in dataTable.Columns)
                {
                    columnMappingsAddMethod.Invoke(columnMappings, new object[] { item.ColumnName, item.ColumnName });//映射字段名称
                }
                try
                {
                    var method = bulkCopyType.GetMethod("WriteToServerAsync", new Type[] { typeof(DataTable) });
                    var writeToServerAsync = (Task)method.Invoke(bulkCopy, new object[] { dataTable });
                    await writeToServerAsync;
                    await tran.CommitAsync();
                    return dataTable.Rows.Count;
                }
                catch
                {
                    if (tran != null)
                    {
                        await tran.RollbackAsync();
                    }
                    throw;
                }
                finally
                {
                    if (tran != null)
                    {
                        await tran.DisposeAsync();
                    }
                    if (bulkCopy != null)
                    {
                        bulkCopyType.GetMethod("Close").Invoke(bulkCopy, null);//关闭
                    }
                    await conn.CloseAsync();
                }
            }
            #endregion

            #region Mysql
            if (ado.DbOptions.DbType == Models.DbType.MySQL)
            {
                var mysqlClient = Assembly.LoadFrom(Path.Combine(AppContext.BaseDirectory, $"{ado.DbOptions.ProviderName}.dll"));
                await conn.OpenAsync();
                var tran = await conn.BeginTransactionAsync();

                var bulkCopy = Activator.CreateInstance(mysqlClient.GetType($"{ado.DbOptions.ProviderName}.MySqlBulkCopy"), new object[]
                {
                    conn,
                    tran
                });

                var bulkCopyType = bulkCopy.GetType();

                bulkCopyType.GetProperty("DestinationTableName").SetValue(bulkCopy, dataTable.TableName);//表名
                bulkCopyType.GetProperty("BulkCopyTimeout").SetValue(bulkCopy, bulkCopyTimeout);

                var columnMappings = bulkCopyType.GetProperty("ColumnMappings").GetValue(bulkCopy);
                var columnMappingType = mysqlClient.GetType($"{ado.DbOptions.ProviderName}.MySqlBulkCopyColumnMapping");
                var columnMappingsAddMethod = bulkCopyType.GetProperty("ColumnMappings").PropertyType.GetMethod("Add", new Type[] { columnMappingType });

                for (int i = 0; i < dataTable.Columns.Count; i++)
                {
                    var mySqlBulkCopyColumnMapping = Activator.CreateInstance(columnMappingType, new object[] { i, dataTable.Columns[i].ColumnName, null });
                    columnMappingsAddMethod.Invoke(columnMappings, new object[] { mySqlBulkCopyColumnMapping });//映射字段名称
                }

                try
                {
                    var method = bulkCopyType.GetMethod("WriteToServer", new Type[] { typeof(DataTable) });
                    var result = method.Invoke(bulkCopy, new object[] { dataTable });
                    await tran.CommitAsync();
                    return dataTable.Rows.Count;
                }
                catch
                {
                    if (tran != null)
                    {
                        await tran.RollbackAsync();
                    }
                    throw;
                }
                finally
                {
                    if (tran != null)
                    {
                        await tran.DisposeAsync();
                    }
                    await conn.CloseAsync();
                }
            }
            #endregion

            #region Oracle
            if (ado.DbOptions.DbType == Models.DbType.Oracle)
            {
                var oracleClient = Assembly.LoadFrom(Path.Combine(AppContext.BaseDirectory, "Oracle.ManagedDataAccess.dll"));
                await conn.OpenAsync();

                var bulkCopy = Activator.CreateInstance(oracleClient.GetType("Oracle.ManagedDataAccess.Client.OracleBulkCopy"), new object[]
                {
                    conn
                });

                var bulkCopyType = bulkCopy.GetType();

                bulkCopyType.GetProperty("DestinationTableName").SetValue(bulkCopy, $"\"{dataTable.TableName}\"");//表名
                bulkCopyType.GetProperty("BulkCopyTimeout").SetValue(bulkCopy, bulkCopyTimeout);

                var columnMappings = bulkCopyType.GetProperty("ColumnMappings").GetValue(bulkCopy);
                var columnMappingsAddMethod = bulkCopyType.GetProperty("ColumnMappings").PropertyType.GetMethod("Add", new Type[] { typeof(string), typeof(string) });
                foreach (DataColumn item in dataTable.Columns)
                {
                    columnMappingsAddMethod.Invoke(columnMappings, new object[] { item.ColumnName, $"\"{item.ColumnName}\"" });//映射字段名称
                }
                try
                {
                    var method = bulkCopyType.GetMethod("WriteToServer", new Type[] { typeof(DataTable) });
                    method.Invoke(bulkCopy, new object[] { dataTable });
                    return dataTable.Rows.Count;
                }
                catch
                {
                    throw;
                }
                finally
                {
                    if (bulkCopy != null)
                    {
                        bulkCopyType.GetMethod("Close").Invoke(bulkCopy, null);//关闭
                    }
                    await conn.CloseAsync();
                }
            }
            #endregion

            #region PostgreSQL
            if (ado.DbOptions.DbType == Models.DbType.PostgreSQL)
            {
                var columns = new List<string>();
                foreach (DataColumn item in dataTable.Columns)
                {
                    columns.Add($"\"{item.ColumnName}\"");
                }
                await conn.OpenAsync();
                var sql = $"COPY \"{dataTable.TableName}\" ( {string.Join(",", columns) } ) FROM STDIN (FORMAT BINARY)";
                var beginBinaryImport = conn.GetType().GetMethod("BeginBinaryImport", new Type[] { typeof(string) });
                var wr = beginBinaryImport.Invoke(conn, new object[] { sql });
                var writeType = wr.GetType();
                var writeRow = writeType.GetMethod("WriteRow", new Type[] { typeof(object[]) });
                try
                {
                    foreach (DataRow item in dataTable.Rows)
                    {
                        writeRow.Invoke(wr, new object[] { item.ItemArray });
                    }
                    writeType.GetMethod("Complete").Invoke(wr, null);//很重要否则不会保存数据到数据库
                    return dataTable.Rows.Count;
                }
                catch
                {
                    throw;
                }
                finally
                {
                    writeType.GetMethod("Close").Invoke(wr, null);
                    await conn.CloseAsync();
                }
            }
            #endregion

            throw new NotSupportedException($"{ado.DbOptions.DbType}不支持的数据库类型");
        }
    }
}
